#include "../include/inherit_socket.h"
#include "message_struct/message.pb.h"
#include "logic_include/logic_macro.h"
#include "logic_include/file_uploader.h"
#include <stdlib.h>
#include <fcntl.h>
#include "../include/my_log.h"
#include <string>
#include "../include/mysql_con_pool.h"
#include "logic_include/file_download.h"
//处理文件上传，下载

//上传
bool inherit_socket::file_upload(connection_p con_p,char * original_msg,char *body, unsigned short body_len)
{
    

    // 根据protobuf的约定，进行反序列化
    file_upload_message *file_upload_msg = new file_upload_message();
    file_upload_msg->ParseFromArray(body, body_len);
    delete[] original_msg; //只要完成序列化，那么就不要了

    //判断这个应该放到哪个线程去处理
    int tar=file_upload_msg->slot()%FILE_UPLOAD_THREAD_NUM;
    file_uploader* loader=file_uploader::get_instance();
    file_uploader_thread_item * item=loader->thread_item_list[tar];
    std::string path_name=file_upload_msg->path()+file_upload_msg->file_name();
    //判断fd_map里有没有
    
    //这个也不需要了
    // auto it=item->fd_map.find(path_name);
    // if(it!=item->fd_map.end()){
    //     it->second->wait_to_write_pkg++;    //反正是原子操作
    // }else{
    //     delete file_upload_msg;
    //     return false;
    // }

    pthread_mutex_lock(&item->mutex);

    // //这里需要判断判断map_fd在不在里边了-----------不需要，这个包能发过来，那必定存在
    // if(item->fd_map.find(path_name)==item->fd_map.end()){
    //     //找不到，说明这个包应该被丢弃
    //     delete file_upload_msg;
    //     pthread_mutex_unlock(&item->mutex);
    //     my_log::get_log_instance()->write_log(LOG_INFO,0,"丢弃一个 %s 的切片",path_name.c_str()); 
    //     return false;
    // }

    // //获取锁后，判断一下文件上传进程是否结束---这个可能是多余的，我会保证上传线程结束前，调用该业务处理函数的线程已经结束，毕竟业务处理是单例
    //再不济，我自己手动处理结束逻辑也可以
    // if(item->this_file_uploader->is_end_file_uploader_thread==FILE_UPLOADER_THREAD_STOP){   
    //     delete file_upload_msg;
    //     pthread_mutex_unlock(&item->mutex);
    //     return false;
    // }

    // if(item->fd_map.find(file_upload_msg->file_name())==item->fd_map.end()){
    //     //没有，那么就创造一个
    //     //先判断是否存在前置文件路径，有直接创建 ----这里加锁了，没事
    //     if(!file_upload_msg->path().empty()){
    //         std::string path="mkdir -p  ./";
    //         path+=file_upload_msg->path();  //这里传过来的路径必定要是 ./aa/bb 的形式
    //         system(path.c_str());
    //     }
    //     std::string file_name="."+file_upload_msg->path()+"/"+file_upload_msg->file_name();
    //     int fd=open(file_name.c_str(),O_CREAT|O_WRONLY,0664);
    //     if(fd==-1){
    //         my_log::get_log_instance()->write_log(LOG_ERR,errno,"文件创建失败");
    //         //这里考虑发送无法上传的消息给客户端----------目前没有处理
    //       
    //         delete file_upload_msg;
    //         pthread_mutex_unlock(&item->mutex);
    //         return false;
    //     }else{
    //         my_log::get_log_instance()->write_log(LOG_INFO,0,"成功创建 文件%s",file_name.c_str());
    //     }
    //     //组装该消息
    //     transforming_file_info file_info;
    //     file_info.fd=fd;
    //     file_info.file_size=file_upload_msg->file_size();
    //     file_info.transformed_bytes=0;
    //     file_info.con_p=con_p;
    //     item->fd_map[file_upload_msg->file_name()]=file_info;   //没有使用移动语义的意义   
    // }
    

    //保证fd_map存在后，往任务队列里添数据---能保证已经存在了，所以上述代码被注释掉了

    item->task_queue.push_back(file_upload_msg);
    pthread_mutex_unlock(&item->mutex);
    
    //在这里修改

    pthread_cond_signal(&item->cond_t);
    
    return true;
}


bool inherit_socket::file_judge_is_continue_to_upload(connection_p con_p,char * original_msg,char * body,unsigned short body_len){
    //这用于判断是否可以断点续传
    //std::cout<<"走到这里"<<std::endl;
    file_judge_is_continue_to_upload_message*  is_continue=new file_judge_is_continue_to_upload_message;
    is_continue->ParseFromArray(body,body_len);
    //只要一prase了，那么就不需要这条原始数据了，直接释放即可
    delete[] original_msg;  
    --con_p->wait_to_deal_received_pkgs;    // 处理掉该连接里的信息
    //组装一下
    std::string path_name=is_continue->path()+is_continue->file_name();
    //去从存放所有可能的中断节点里判断是否存在
    std::map<std::string,char> *all_info_fd_map=&file_uploader::get_instance()->all_info_fd_map;    //取地址，这样才能更改本体

    auto it=all_info_fd_map->find(path_name);    //根据文件名去查找，是否存在与某个处理线程里
    
    response_of_is_continue_to_upload_message respons_of_is_continue;   //后续用作响应给客户端的信息
    if(it==all_info_fd_map->end()){
        //这是一个全新的待传输的文件
        if(is_continue->path()!="./"){
            //那么需要创建这个路径
            std::string command;
            command="mkdir -p ./"+is_continue->path();
            system(command.c_str());
        }
        

        //然后去创建这个文件。创建这个文件夹的前缀，用用户名来作为唯一标识
        std::string need_create_path_name="./"+path_name;
        int new_fd=open(need_create_path_name.c_str(),O_CREAT|O_RDWR,0664);
        
        if(new_fd==-1){
            //这个逻辑目前并不完善--如果创建失败，那么其实可以去去除掉无效的fd,比如中断里的fd
            my_log::get_log_instance()->write_log(LOG_ERR,errno,"文件 %s 创建失败！！ ",path_name.c_str());
            delete is_continue;
            respons_of_is_continue.set_is_enable_send(false);
        }else{
            //文件创建成功了
            //我应该去判断文件是否已经存在，避免文件重复传，浪费带宽--------------
            long long lseek_res=lseek(new_fd,0,SEEK_END);
            bool is_retransfer=0;
            if(lseek_res){
                if(lseek_res!=is_continue->file_size()){
                    my_log::get_log_instance()->write_log(LOG_ALERT,0,"见鬼了，文件 %s 怎么可能不相等，那么",path_name.c_str());
                    is_retransfer=1;
                    //可能经过一次服务器关机后，之传输了一部分
                }
                
                //如果不需要重传，那么意味着传好了
                if(!is_retransfer){
                    respons_of_is_continue.set_is_enable_send(true);
                    respons_of_is_continue.set_path(is_continue->path());
                    respons_of_is_continue.set_file_name(is_continue->file_name());
                    respons_of_is_continue.set_is_break(false);
                    respons_of_is_continue.set_file_is_exists(1);   // 文件已经存在
                    close(new_fd);  //把它关了
                }
                
            }else{
                transforming_file_info *new_map_item =new transforming_file_info();
                new_map_item->fd=new_fd;
                new_map_item->file_size=is_continue->file_size();
                new_map_item->con_p=con_p;
                new_map_item->transformed_bytes=0;
                new_map_item->icurrentsequence=con_p->icurrentsequence;
                
                //根据slot,找到目标
                int tar_thread_item=is_continue->slot()%FILE_UPLOAD_THREAD_NUM;
                file_uploader_thread_item * item=file_uploader::get_instance()->thread_item_list[tar_thread_item];
                //更新全局文件传输信息表
                all_info_fd_map->insert(std::make_pair(path_name,tar_thread_item));
                //这还是要上锁滴
                pthread_mutex_lock(&item->mutex);
                item->fd_map[path_name]=new_map_item;
                
                pthread_mutex_unlock(&item->mutex);
                //放入成功
                respons_of_is_continue.set_is_enable_send(true);
                respons_of_is_continue.set_path(is_continue->path());
                respons_of_is_continue.set_file_name(is_continue->file_name());
                respons_of_is_continue.set_is_break(false);
                respons_of_is_continue.set_slot(is_continue->slot());
                respons_of_is_continue.set_file_is_exists(false);
                //写个日志吧
                my_log::get_log_instance()->write_log(LOG_INFO,0,"成功创建一个待上传文件 %s 到fd的映射,并放入指定位置",path_name.c_str());
            }
            //如果没传完，服务器宕机，那么需要重传
            if(is_retransfer){
                transforming_file_info *new_map_item =new transforming_file_info();
                new_map_item->fd=new_fd;
                new_map_item->file_size=is_continue->file_size();
                new_map_item->con_p=con_p;
                new_map_item->transformed_bytes=0;
                new_map_item->icurrentsequence=con_p->icurrentsequence;
                
                //根据slot,找到目标
                int tar_thread_item=is_continue->slot()%FILE_UPLOAD_THREAD_NUM;
                file_uploader_thread_item * item=file_uploader::get_instance()->thread_item_list[tar_thread_item];
                //更新全局文件传输信息表
                all_info_fd_map->insert(std::make_pair(path_name,tar_thread_item));
                //这还是要上锁滴
                pthread_mutex_lock(&item->mutex);
                item->fd_map[path_name]=new_map_item;
                
                pthread_mutex_unlock(&item->mutex);
                //放入成功
                respons_of_is_continue.set_is_enable_send(true);
                respons_of_is_continue.set_path(is_continue->path());
                respons_of_is_continue.set_file_name(is_continue->file_name());
                respons_of_is_continue.set_is_break(false);
                respons_of_is_continue.set_slot(is_continue->slot());
                respons_of_is_continue.set_file_is_exists(false);
                //写个日志吧
                my_log::get_log_instance()->write_log(LOG_INFO,0,"重传----成功创建一个待上传文件 %s 到fd的映射,并放入指定位置",path_name.c_str());
            }

        }

    }else{
        //说明全局中有，那么考虑发送断点---下面这个逻辑存在线程安全问题，
        
        char tar_thread_item=it->second;    //得到所在线程item--------1
        file_uploader_thread_item * item=file_uploader::get_instance()->thread_item_list[tar_thread_item];

        
        
        
        //这里判断发送过来的包，是否全部处理掉了
        //拿到con_p与icurrentsequence
        connection_p old_con_p=nullptr;
        int old_icurrentsequence=0;
        auto fd_map_item=item->fd_map.find(path_name); //-------------2
        if(fd_map_item!=item->fd_map.end()){
            //存在
            old_con_p=fd_map_item->second->con_p;
            old_icurrentsequence=fd_map_item->second->icurrentsequence;

        }

        while(old_con_p!=nullptr&&!(old_con_p->wait_to_deal_received_pkgs==0&& (old_icurrentsequence!=old_con_p->icurrentsequence))){
            //如果没有要处理的包,且迭代器还有效
            sleep(1);   //睡一秒
        }
        //只要走到这里，就能够保证包全处理掉了
        
        //首先判断文件是否是已经传输完成的
        if(item->fd_map.find(path_name)==item->fd_map.end()){
            //只需再找一次，毕竟如果已经传输完毕，那么就会被erase掉
            //这里表明文件传输完毕
            respons_of_is_continue.set_is_enable_send(true);
            respons_of_is_continue.set_path(is_continue->path());
            respons_of_is_continue.set_file_name(is_continue->file_name());
            respons_of_is_continue.set_is_break(false);
            respons_of_is_continue.set_file_is_exists(1);   // 文件已经存在
            respons_of_is_continue.set_slot(tar_thread_item);


        }else{
            
             

            //更新
            fd_map_item->second->con_p=con_p;
            fd_map_item->second->icurrentsequence=con_p->icurrentsequence;
        

            //那么发送对应文件的断点给客户端
            respons_of_is_continue.set_is_enable_send(true);
            respons_of_is_continue.set_path(is_continue->path());
            respons_of_is_continue.set_file_name(is_continue->file_name());
            respons_of_is_continue.set_is_break(true);
            respons_of_is_continue.set_break_info(fd_map_item->second->transformed_bytes);  //发字节就可以了
            respons_of_is_continue.set_slot(tar_thread_item);
            respons_of_is_continue.set_file_is_exists(false);
        
            
            my_log::get_log_instance()->write_log(LOG_INFO,0,"成功将断点的文件%s 的信息 文件发给客户端 其中断点=%d ",path_name.c_str(),fd_map_item->second->transformed_bytes);

        }

        
       
        
    }
    //最后都需要将这个信息发送出去
    std::cout<<"走到这里"<<std::endl;
    std::string true_response_data;
    respons_of_is_continue.SerializeToString(&true_response_data);
    move_message_to_send_list(con_p,true_response_data.c_str(),true_response_data.length(),RESPONSE_OF_JUDGE_IS_CONTINUE_TO_UPLOAD);
    std::cout<<"返回来了"<<"本次组装长度为："<<true_response_data.length()<<std::endl;
    return true;
    //后面这个true_response_data自动消亡
}

//下载

//处理发送过来的下载请求
bool inherit_socket::file_judge_is_continue_to_download(connection_p con_p,char * original_msg,char * body,unsigned short body_len){
    
    //反序列化得到数据-----这几行，无论是什么业务处理，都是雷打不动的
    file_judge_is_continue_to_download_message * download_req=new file_judge_is_continue_to_download_message;
    download_req->ParseFromArray(body,body_len);
    delete original_msg;
    --con_p->wait_to_deal_received_pkgs;    //减少该连接堆积的请求消息

    std::string client_req=download_req->file_path_name();  //组装sql语句时会加上通配符
    //构建这个sql语句
    char sql[512];
    memset(sql,0,512);
    snprintf(sql,512,"select file_client_filename,file_server_filename,file_size from file where file_client_filename like '%s' ",(client_req+"%").c_str());

    MYSQL * now_con=mysql_con_pool::get_instance()->get_con();
    //如果返回NULL也得有一套逻辑兜底，目前暂时不考虑-----------可以考虑等待一定的时间，再度尝试获取
    if(now_con==nullptr){
        perror("在file_judge_is_continue_to_download函数中，获取mysql连接失败：");
        delete download_req;
        return false;
        //这里应该考虑发送下载失败的回应
    }

    MYSQL_RES* res=mysql_con_pool::get_instance()->get_query_result_by_use(now_con,sql);
    if(res==NULL){
        my_log::get_log_instance()->write_log(LOG_INFO,0,"客户端文件下载请求 %s,没有查询出数据",client_req.c_str());
        mysql_con_pool::get_instance()->return_con(now_con);
        delete download_req;
        return false;
    }
    //如何到达目标线程--暂时就先采用work_id实现负载均衡吧
    file_downloader_thread_item * thread_item=file_downloader::get_instance()->get_one_thread_item(download_req->work_id());
    
    //下载请求任务的单位类
    work_item * work_task=new work_item();
    work_task->con_p=con_p;
    work_task->client_request=client_req;
    work_task->work_id=download_req->work_id();
    work_task->icurrentsequence=con_p->icurrentsequence;
    
    MYSQL_ROW row;  //用于拿取每一行数据
    file_item * now_file_item;  //任务中具体到的每个文件任务
    long long now_file_size=0;
    file_download_spy_message to_client_spy_message;        // 待发送的数据
    to_client_spy_message.set_work_id(download_req->work_id());  //继续把work_id返还给客户端
    while(row=mysql_fetch_row(res)){
        //循环拿取符合条件的数据
        now_file_item=new file_item();

        now_file_item->file_servername=row[1];
        now_file_item->opened_fd=-1;
        now_file_size=atoll(row[2]);
        now_file_item->file_size=now_file_size;
        
        //这里也需要组装刺探消息发送给客户端
        
        to_client_spy_message.add_path_file_name(row[0]);

        to_client_spy_message.add_file_size(now_file_size);
        
        work_task->work_item_map.insert(std::make_pair(row[0],now_file_item));
        //test -----------------看查询到的数据
        std::cout<<row[0]<<" "<<row[1]<<" "<<row[2]<<std::endl;

    }

    //已经全部放入，并都发送数据给客户端了,这里应该要上锁互斥
    pthread_mutex_lock(&thread_item->tmp_task_map_mutex);
    thread_item->tmp_task_map.insert(std::make_pair(std::make_pair(con_p,download_req->work_id()),work_task));
    pthread_mutex_unlock(&thread_item->tmp_task_map_mutex);

    
    std::string to_client_data;
    to_client_spy_message.SerializeToString(&to_client_data);
    move_message_to_send_list(con_p,to_client_data.c_str(),to_client_data.length(),FILE_DOWNLOAD_SPY);

    

    //归还mysql_连接
    mysql_free_result(res); //清空数据
    mysql_con_pool::get_instance()->return_con(now_con);
    
    delete download_req;
    
    return true;
    


    
}

//客户端接收到spy数据后回复给服务器的响应数据
bool inherit_socket::response_of_is_continue_to_download(connection_p con_p, char *original_msg, char *body, unsigned short body_len)
{
    //反序列化得到数据
    response_of_download_spy_message *res_of_spy_msg=new response_of_download_spy_message();
    res_of_spy_msg->ParseFromArray(body,body_len);
    delete original_msg;
    --con_p->wait_to_deal_received_pkgs;    //减少待处理的堆积的消息

    //通过work_id找到对应的工作线程file_downloader_thread_item *
    file_downloader_thread_item * thread_item=file_downloader::get_instance()->get_one_thread_item(res_of_spy_msg->work_id());
    
    //再通过con_p与work_id的pair找到对应的具体的任务体work_item   
    work_item * work_task=nullptr;  //=thread_item->task_map[std::make_pair(con_p,res_of_spy_msg->work_id())];
    auto task_map_it=thread_item->tmp_task_map.find(std::make_pair(con_p,res_of_spy_msg->work_id()));

    //未处理响应前只能在tmp_task_map--------------
    if(task_map_it==thread_item->tmp_task_map.end()){
        
        //这里如果还没有找到，那么奇怪了，写日志记录一下
        my_log::get_log_instance()->write_log(LOG_ERR,0,"服务器接收到的下载刺探数据的回应居然在对应的线程item里找不到");
        delete res_of_spy_msg;
        return false;
    }else{
        work_task=task_map_it->second;
    }
    //由于对于客户端来说的每个下载任务,可能有多个文件需要发送，在work_task中找到对应的具体的文件下载请求，更新状态
    
    std::string tmp_file_path_name; //暂存发送过来的文件路径
    std::map<std::string,file_item*>::iterator work_item_map_it;
    
    for(int i=0;i<res_of_spy_msg->path_file_name_size();++i){
        tmp_file_path_name=res_of_spy_msg->path_file_name(i);
        work_item_map_it=work_task->work_item_map.find(tmp_file_path_name);
        
        if(work_item_map_it==work_task->work_item_map.end()){
            //应该不会走到这里，如果走到这里，那么写个日志记录一下
            my_log::get_log_instance()->write_log(LOG_ERR,0,"居然在work_item_map里找不到发送过来的文件路径的映射！！！");
            break;
        }
        //如果是已经传输完毕的，那么要移除掉
        work_item_map_it->second->transformed_size=res_of_spy_msg->transformed_size(i);
        if(work_item_map_it->second->transformed_size==work_item_map_it->second->file_size){
            //说明是是已经传输完成的，那么移除该file_item即可
            delete work_item_map_it->second;
            work_task->work_item_map.erase(work_item_map_it);
            continue;
        }
        work_item_map_it->second->seq=res_of_spy_msg->transformed_size(i)/1024; //由于是以1024为基本单位发送的  序列号是以0开始发送的，
                                                                                //如果是1024，那么seq就为1，下一次发送的seq也为1                                                                        
    }

    if(work_task->work_item_map.empty()){
        //已经为空了，那么移除task_map_it
        delete work_task;
        pthread_mutex_lock(&thread_item->tmp_task_map_mutex);
        thread_item->tmp_task_map.erase(task_map_it);
        pthread_mutex_unlock(&thread_item->tmp_task_map_mutex);
        delete res_of_spy_msg;
        return true;
        
    }
    work_task->is_active=WORK_ITEM_ACTIVE;  //这个参数已经没什么用了

    //这里的加入操作可能会由于进行i/o的发送而阻塞较长时间，那只能说响应慢一点，但线程安全的
    pthread_mutex_lock(&thread_item->mutex);
    //将当前的活跃item放入即可
    thread_item->task_map.insert((*task_map_it));

    pthread_mutex_unlock(&thread_item->mutex);
    
    pthread_cond_signal(&thread_item->cond);    //唤醒一下
    std::cout<<"接收到刺探报文的回应了，正式通知线程去发送了"<<std::endl;

    //
    pthread_mutex_lock(&thread_item->tmp_task_map_mutex);
    thread_item->tmp_task_map.erase(task_map_it);   //移除
    pthread_mutex_unlock(&thread_item->tmp_task_map_mutex);

    delete res_of_spy_msg;
    return true;
}